package jp.co.tms.internal.socket;

import java.io.IOException;
import java.net.BindException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;

import jp.co.tms.internal.runserver.exception.ConnectionIsInUseException;
import jp.co.tms.internal.runserver.exception.ConnectionIsOutException;
import jp.co.tms.internal.runserver.exception.PortNumberIsUnLegalException;
import jp.co.tms.internal.runserver.protocol.IPackage;

public class RunServerSocket extends Thread{
	// NIO elements.
	private ISocketEventHandle _worker;
	private ServerSocketChannel _channel;
	private Selector _selector;
	// the end flag.
	private boolean _flag = false;
	// database
	private Map<SocketChannel, List> _model = new HashMap<SocketChannel, List>();
	// change ops event list
	private Queue<ChangeOps> _queue = new LinkedList<ChangeOps>();
	
	public RunServerSocket(ISocketEventHandle handle){	
		this._worker = handle;
	}
	
	/**
	 * 
	 * @param port
	 * @return
	 * @throws ConnectionIsInUseException
	 * @throws PortNumberIsUnLegalException
	 */
	public void bind(int port) throws
	ConnectionIsInUseException,
	PortNumberIsUnLegalException,
	BindException,
	IOException
	{
		//
		// If the connection is in use.
		if (_flag){
			throw new ConnectionIsInUseException();
		}
	
		//
		// if the port is not legel.
		if (port <= 256 || port >= 65535){
			throw new PortNumberIsUnLegalException();
		}
		
		//
		// bind the port.
		// open the port and bind the socket to it.
		_channel = ServerSocketChannel.open() ;
		ServerSocket ss = _channel.socket() ;
		InetSocketAddress address = new InetSocketAddress(port) ; 
		ss.bind(address) ;
		
		// add the selector to socket.
		_channel.configureBlocking(false) ;
		_selector = Selector.open() ;
		_channel.register(_selector, SelectionKey.OP_ACCEPT) ;
		
		_flag = true;
	}
	
	/**
	 * 
	 */
	public void listen() throws
	ConnectionIsOutException
	{
		if (!_flag){
			throw new ConnectionIsOutException();
		}
		
		this.start();
	}
	
	public void run(){
		//
		// check the selector in loop.
		while(_flag){
			try {
				
				synchronized (_queue){
					//
					// set the channel to write.
					while (true){
						ChangeOps change = _queue.poll();
						if (change == null)
							break;
						
						SelectionKey key = change.socket.keyFor(_selector);
						key.interestOps(SelectionKey.OP_WRITE);
					}
				}
				//
				// select.
		        _selector.select();
	
				Set<SelectionKey> readyKeys = _selector.selectedKeys();
				Iterator<SelectionKey> iter = readyKeys.iterator();
				while(iter.hasNext()){
				    SelectionKey key = iter.next();
				    iter.remove();
				    
				    if (!key.isValid()){
				    	continue;
				    }
				    
					if(key.isAcceptable()){ 
						accept(key);
					}
					else if(key.isReadable()){
						read(key);
					}
					else if(key.isWritable()){			
						write(key);
					}
					else{
					}
				}
			}
			catch(IOException e){
				e.printStackTrace();
			}
		}
	}
	
	@SuppressWarnings({ "rawtypes", "unchecked" })
	public void send(SocketChannel sc, byte[] data){
		// if the sending data is null, just return.
		if (data == null)
			return;
		
		synchronized (_queue){
			//
			// add the buffer to queue.
			_queue.add(new ChangeOps(sc, SelectionKey.OP_WRITE));
			//
			// if there are no key in the map, create one.
			if (!_model.containsKey(sc)){
				_model.put(sc, new LinkedList<byte[]>());
			}
			//
			// add the data to list.
			List list = _model.get(sc);
			list.add(data);
			//
			// wakeup the selector to set WRITE ops to channel. 
			_selector.wakeup();
		}
	}
	
	public void close(){	
		_flag = false;
	}
	
	protected void disconnect(SocketChannel sc){
		// if there are some data in pending area, remove it.
		if (_model.containsKey(sc)){
			_model.remove(sc);
		}
		
		// if there are change ops data in list, remove it.
		Iterator<ChangeOps>  iterator = _queue.iterator();
		while (iterator.hasNext()){
			ChangeOps change = iterator.next();
			if (sc.equals(change.socket)){
				_queue.remove(change);
			}
		}
		
		// notify the user.
		_worker.disconnect(sc);
	}
	
	protected void accept(SelectionKey key) throws IOException{
		ServerSocketChannel server = (ServerSocketChannel) key.channel();
		SocketChannel sc = server.accept();
		sc.configureBlocking(false);
		sc.register(_selector, SelectionKey.OP_READ);
		// Tell the user to setup the database.
		_worker.connect(sc);	
	}
	
	protected void read(SelectionKey key) throws IOException{
		SocketChannel sc = (SocketChannel) key.channel();
		try {
			sc.configureBlocking(false);
			ByteBuffer buffer = ByteBuffer.allocate(IPackage.PACKAGE_LEN);
			int numRead = sc.read(buffer);
			if ( numRead == -1 ){
				// close the channel.
				key.channel().close();
				key.cancel();
				// tell the user to set the disconnect event.
				disconnect(sc);
			}
			else if(numRead == 0){
				return;
			}

			buffer.flip();
			// call the handle when the data has been received.
			_worker.processData(sc, buffer.array());
		}
		catch (IOException e) {
			// close the channel.
			key.cancel();
		    sc.close();
		    // tell the user to set the disconnect event.
		    disconnect(sc);
		}		
	}
	
	@SuppressWarnings("rawtypes")
	protected void write(SelectionKey key) throws IOException{
		SocketChannel sc = (SocketChannel) key.channel();
		List list = _model.get(sc);
		if (list == null){
			return;
		}
		
		Iterator iterator = list.iterator();
		while (iterator.hasNext()){
			byte[] buf = (byte[])iterator.next();
			ByteBuffer buffer = ByteBuffer.wrap(buf);
			sc.write(buffer);
			if (buffer.remaining() > 0){
				break;
			}
			// if it success to send, remove it from the list.
			list.remove(buf);
		}
		
		// if all the data has been sent, set the channel back to READ_ONLY.
		if (list.isEmpty()){
			key.interestOps(SelectionKey.OP_READ);
		}
	}
}
